package com.niodata.md.rdb.crawler;

import com.niodata.md.rdb.model.DataBase;
import com.niodata.md.rdb.model.Table;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Collections;
import java.util.List;
import org.apache.commons.lang3.ArrayUtils;

/**
 * retrieve table/view create sql.
 */
public class CreateSqlRetrieval extends AbstractRetrieval {
  @Override
  public List<MetaDataRetrieval> retrieveMetaData() throws Exception {
    DataBase dataBase = this.getCatalog().getDataBase(
        ArrayUtils.subarray(this.getParentPath(), 0, 2));
    Table table = this.getCatalog().getTable(this.getParentPath());
    Connection connection = this.getDataSourceInfo().getConnection(dataBase.getName(), "");
    try {
      String sql = this.getCreateSql(table.getDatabase(), table.getSchema(), table.getName());
      Statement statement = connection.createStatement();
      ResultSet rs = statement.executeQuery(sql);
      rs.next();
      int index = this.getResultColumnIndex();
      table.setCreateSql(rs.getString(index));
      return Collections.emptyList();
    } finally {
      connection.close();
    }
  }

  private int getResultColumnIndex() {
    switch (this.getCatalog().getDbInfo().getDatabaseServerType()) {
      case "mysql", "doris":
        return 2;
      case "postgresql":
        return 1;
      default:
        return -1;
    }
  }

  private String getCreateSql(String dbName, String schemaName, String tableName) {
    switch (this.getCatalog().getDbInfo().getDatabaseServerType()) {
      case "mysql", "doris":
        return "show create table `" + tableName + "`";
      case "postgresql":
        return getPostgreDescTable(dbName, schemaName, tableName);
      case "sqlserver":
        return getSqlserverDescTable(dbName, tableName);
      default:
        return "";
    }
  }

  /**
   * 获取 SQL Server 描述表结构的 SQL 语句.
   *
   * @param dbName    数据库名称
   * @param tableName 表名称
   * @return 描述表结构的 SQL 语句
   */
  private static String getSqlserverDescTable(String dbName, String tableName) {
    return String
        .format("use %s"
            + "\n"
            + "SELECT \n"
            + "        col.name AS  COLUMN_NAME,\n"
            + "        t.name AS COLUMN_TYPE ,\n"
            + "        CASE WHEN EXISTS ( SELECT   1\n"
            + "                           FROM     dbo.sysindexes si\n"
            + "                                 INNER JOIN dbo.sysindexkeys sik ON si.id = sik.id\n"
            + "                                                          AND si.indid = sik.indid\n"
            + "                                 INNER JOIN dbo.syscolumns sc ON sc.id = sik.id\n"
            + "                                                          AND sc.colid = sik.colid\n"
            + "                                 INNER JOIN dbo.sysobjects so ON so.name = si.name\n"
            + "                                                          AND so.xtype = 'PK'\n"
            + "                           WHERE    sc.id = col.id\n"
            + "                                    AND sc.colid = col.colid ) THEN 'PRI'\n"
            + "             ELSE ''\n"
            + "        END AS COLUMN_KEY ,\n"
            + "        CONVERT(varchar(200), ISNULL(ep.[value], '')) AS COMMENT \n"
            + "  \n"
            + "FROM    dbo.syscolumns col\n"
            + "        LEFT  JOIN dbo.systypes t ON col.xtype = t.xusertype\n"
            + "        inner JOIN dbo.sysobjects obj ON col.id = obj.id\n"
            + "                                         AND obj.xtype = 'U'\n"
            + "                                         AND obj.status >= 0\n"
            + "        LEFT  JOIN dbo.syscomments comm ON col.cdefault = comm.id\n"
            + "        LEFT  JOIN sys.extended_properties ep ON col.id = ep.major_id\n"
            + "                                         AND col.colid = ep.minor_id\n"
            + "                                         AND ep.name = 'MS_Description'\n"
            + "        LEFT  JOIN sys.extended_properties epTwo ON obj.id = epTwo.major_id\n"
            + "                                         AND epTwo.minor_id = 0\n"
            + "                                         AND epTwo.name = 'MS_Description'\n"
            + "WHERE   obj.name = '%s'\n"
            + "ORDER BY col.colorder ;\n", dbName, tableName);
  }

  /**
   * 获取 PostgreSQL 描述表结构的 SQL 语句.
   *
   * @param dbName     数据库名称
   * @param schemaName 模式名称
   * @param tableName  表名称
   * @return 描述表结构的 SQL 语句
   */
  private static String getPostgreDescTable(String dbName, String schemaName, String tableName) {
    return String
        .format("WITH d AS\n"  //schema
                + "  (SELECT a.attname AS COLUMN,\n"
                + "          format_type(a.atttypid, a.atttypmod) AS data_type,\n"
                + "          'PRI' AS column_key\n"
                + "   FROM pg_index i\n"
                + "   JOIN pg_attribute a ON a.attrelid = i.indrelid\n"
                + "   AND a.attnum = ANY(i.indkey)\n"
                + "   WHERE i.indrelid = '%s.%s'::regclass\n" //schema.table
                + "     AND i.indisprimary),\n"
                + "     c AS\n"
                + "  (SELECT a.*,\n"
                + "          b.comment\n"
                + "   FROM\n"
                + "     (SELECT COLUMN_NAME AS COLUMN,\n"
                + "                            CASE data_type\n"
                + "                                WHEN 'character' THEN 'char'\n"
                + "                                WHEN 'int4' THEN 'integer'\n"
                + "                                WHEN 'int2' THEN 'smallint'\n"
                + "                                WHEN 'int8' THEN 'bigint'\n"
                + "                                WHEN 'double precision' THEN 'double'\n"
                + "                                WHEN 'time without time zone' THEN 'time'\n"
                + "                                WHEN 'timestamp without time zone' THEN 'timestamp'\n"
                + "                                WHEN 'character varying' THEN 'varchar'\n"
                + "                                WHEN 'numeric' THEN 'numeric'\n"
                + "                                ELSE data_type\n"
                + "                            END AS TYPE\n"
                + "      FROM information_schema.columns\n"
                + "      WHERE table_schema = '%s'\n" // schema
                + "        AND TABLE_NAME = '%s') a,\n" // table
                + "\n"
                + "     (SELECT cols.column_name AS COLUMN,\n"
                + "\n"
                + "        (SELECT pg_catalog.col_description(c.oid, cols.ordinal_position::int)\n"
                + "         FROM pg_catalog.pg_class c\n"
                + "         WHERE c.oid =\n"
                + "             (SELECT cols.table_name::regclass::oid)\n"
                + "           AND c.relname = cols.table_name ) AS COMMENT\n"
                + "      FROM information_schema.columns cols\n"
                + "      WHERE cols.table_catalog = '%s'\n" // dbname (pg datasource name)
                + "        AND cols.table_schema = '%s'\n" // schema name
                + "        AND cols.table_name = '%s') b\n" // table name
                + "   WHERE a.column=b.column )\n"
                + "SELECT c.*,\n"
                + "       d.column_key\n"
                + "FROM c\n"
                + "LEFT JOIN d ON c.column=d.column", schemaName, tableName, schemaName, tableName, dbName,
            schemaName, tableName);
  }

}
